package org.redis;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.springframework.data.redis.connection.*;
import org.springframework.data.redis.core.*;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.DefaultScriptExecutor;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.stereotype.Service;

import java.io.Closeable;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * redis 监控工具类
 */
@Service
public class RedisAnalyzerService implements Closeable {

    private final RedisTemplate<String,Object> template;
    private  RedisConnection connection;
    private final MyScriptExecutor scriptExecutor;

    public RedisAnalyzerService(RedisTemplate<String,Object> template){
        this.template = template;
        this.scriptExecutor = new MyScriptExecutor(template);
        this.connection = getConnection();
    }

    private RedisConnection getConnection(){
        if(checkConnection()) return this.connection;
        return this.connection=RedisConnectionUtils.getConnection(getRedis().getRequiredConnectionFactory());
    }
    /*
    *
    *
    * */

    private MyScriptExecutor getExecutor(){return this.scriptExecutor;}

    private RedisTemplate<String,Object> getRedis(){return this.template;}

    private boolean checkConnection(){
        return !Objects.isNull(connection)&&!connection.isClosed();
    }

    private String execCommand(RedisInfo info){
        RedisConnection connection = getConnection();
            if(connection instanceof RedisClusterConnection){
                RedisClusterConnection redisClusterConnection = (RedisClusterConnection) connection;
                Iterable<RedisClusterNode> redisClusterNodes = redisClusterConnection.clusterGetNodes();
                Iterator<RedisClusterNode> iterator = redisClusterNodes.iterator();
                Map<String,String> valueMap = new HashMap<String,String>(redisClusterConnection.clusterGetClusterInfo().getClusterSize().intValue());
                while(iterator.hasNext()){
                    RedisClusterNode next = iterator.next();
                    String propertyName = String.format("%s:%d.%s", next.getHost(), next.getPort(), info.getCommand());
                    String property = redisClusterConnection.info(info.getSection()).getProperty(propertyName,"0");
                    valueMap.put(propertyName,property);
                }
                return JSON.toJSONString(valueMap);
            }
            final Properties properties = connection.info(info.getSection());
            return properties != null?properties.getProperty(info.getCommand(),"0"):"not found";

     }

     public void execute(){

         final RedisConnection connection = getConnection();

//         final RedisCommand scan = RedisCommand.SCAN;
//         scan. 0 match key* count 100
         //
         /**
          *
          * 集群环境下如何使用lua。需要使用内置的方法，计算hash值到具体的节点
          *
          * local key = KEYS[1]
          * local value = ARGV[1]
          *
          * -- 计算键的哈希值
          * local hash = redis.call('CRC16',key)
          * -- 计算分片节点编号 NUM_OF_SHARDS 是你的自己配置了多少，例如16484
          * local shard = hash%NUM_OF_SHARDS
          *
          * -- 发送操作到对于的分片节点
          * redis.call('SELECT',shard)
          * redis.call('SET',key,value )
          *
          * 此外lua脚本中，不能操作多个键，例如mget，mset等
          *
          */
         connection.execute("scan","0".getBytes(StandardCharsets.UTF_8),
                 "keys*".getBytes(StandardCharsets.UTF_8),"100".getBytes(StandardCharsets.UTF_8));

     }

    public String memoryUsage(){
        return execCommand(RedisInfo.MEMORY_USAGE);
    }

    /**
     * 获取已使用内存，单位是bytes
     * @param human 是否以人可读的方式
     * @return 具体的字节数
     */
    public String redisMemory(boolean human){
        return human?execCommand(RedisInfo.USE_MEMORY_HUMAN):execCommand(RedisInfo.USE_MEMORY);
    }
    public String redisMemory(){
        return redisMemory(false);
    }

    public String osMemory(){
        return osMemory(false);
    }

    /**
     * 以操作系统视角查看redis 占用
     */
    public String osMemory(boolean human){
        return human?execCommand(RedisInfo.USE_MEMORY_RSS_HUMAN):execCommand(RedisInfo.USE_MEMORY_RSS);
    }

    /**
     * redis数据占用
     */
    public String dataMemory(){
        return execCommand(RedisInfo.USE_MEMORY_DATASET);
    }

    /**
     * used_memory_dataset在净内存（used_memory-used_memory_startup）使用量中所占的百分比
     */
    public String dataMemoryPercent(){
        return execCommand(RedisInfo.USE_MEMORY_DATASET_PERC);
    }

    /**
     * 本机器的总内存
     *
     */
    public String totalMemory(boolean human){
        return human?execCommand(RedisInfo.TOTAL_SYSTEM_MEMORY_HUMAN):execCommand(RedisInfo.TOTAL_SYSTEM_MEMORY);
    }
    public String totalMemory(){
        return totalMemory(false);
    }

    /**
     * 获得cpu使用率
     *
     */
//    public String cpuPercent(){
//        final String cpuSys = execCommand(RedisInfo.USED_CPU_SYS);
//        final String cpuUser = execCommand(RedisInfo.USED_CPU_USER);
//        return String.format("%s",Float.parseFloat(cpuSys)+Float.parseFloat(cpuUser));
//    }

    private String cpuInfo(){
        final String cpuSys = execCommand(RedisInfo.USED_CPU_SYS);
        RedisInfo usedCpuUser = RedisInfo.USED_CPU_USER;
        String command = usedCpuUser.getCommand();
        final String cpuUser = execCommand(usedCpuUser);
        // 可能存在分布式，查看是否需要转换
        try {
            JSONObject cpuSysObject = JSON.parseObject(cpuSys);
            JSONObject cpuUserObject = JSON.parseObject(cpuUser);
            Set<Map.Entry<String, Object>> entries = cpuSysObject.entrySet();
            Map<String,String> result = new HashMap<>(entries.size());
            for(Map.Entry<String,Object> entry:cpuSysObject.entrySet()){
                String key = entry.getKey();
                String hostPort = key.substring(0,key.lastIndexOf("."));
                String userKey = String.format("%s.%s", hostPort, command);
                String newKey = String.format("%s.cup_percent", hostPort);
                if(cpuUserObject.containsKey(userKey)){
                    result.put(newKey,String.format("%f",Float.parseFloat(cpuSysObject.getString(key))+
                            Float.parseFloat(cpuUserObject.getString(userKey))));
                }
            }
            return JSON.toJSONString(result);
        }catch (Exception e){e.printStackTrace(); }
        return String.format("%s",Float.parseFloat(cpuSys)+Float.parseFloat(cpuUser));
    }

    /**
     * 获得cpu使用率
     *
     */
    public String cpuPercent(){
        String first = "";
        String last = "";
        long firstTime = System.currentTimeMillis();
        first = cpuInfo();
        last = cpuInfo();
        float diffTime = (System.currentTimeMillis() - firstTime)/1_000f;//转换为秒
        try {
            JSONObject firstJson = JSON.parseObject(first);
            JSONObject lastJson = JSON.parseObject(last);
            Set<Map.Entry<String, Object>> entries = firstJson.entrySet();
            Map<String,String> result = new HashMap<>(entries.size());
            for(Map.Entry<String,Object> entry:firstJson.entrySet()){
                String key = entry.getKey();
                if(lastJson.containsKey(key)){
                    float abs = Math.abs(Float.parseFloat(firstJson.getString(key)) - Float.parseFloat(lastJson.getString(key)));
                    result.put(key,String.format("%.2f",(abs/diffTime)*100));
                }
            }
            return JSON.toJSONString(result);
        }catch (Exception ignore){ }
        float abs = Math.abs(Float.parseFloat(first) - Float.parseFloat(last));
        return String.format("%.2f",(abs/diffTime)*100);
    }



    /**
     * 获得列表key的数量
     *
     */
    public Map<String,Long> keyCountCluster(String keyPattern) {
        RedisConnection connection = getConnection();
            if(connection instanceof RedisClusterConnection){
                RedisClusterConnection redisClusterConnection = (RedisClusterConnection) connection;
                redisClusterConnection.execute("");
                Iterable<RedisClusterNode> redisClusterNodes = redisClusterConnection.clusterGetNodes();
                Map<String,Long> result = new HashMap<>(redisClusterConnection.clusterGetClusterInfo().getClusterSize().intValue());
                ScanOptions scanOptions = ScanOptions.scanOptions().match(keyPattern).count(Integer.MAX_VALUE).build();
                for (RedisClusterNode redisClusterNode : redisClusterNodes) {
//                    List<String> list = new ArrayList<>();
                    AtomicLong along = new AtomicLong(0);
                    try (Cursor<byte[]> scan = redisClusterConnection.scan(redisClusterNode, scanOptions)){
                        while(scan.hasNext()){
                            scan.next();
                            along.incrementAndGet();
                            // 不获取值
//                            list.add(new String(scan.next()));
                        }
                    }catch (Exception ignore){}
                    result.put(String.format("%s:%s.%s",redisClusterNode.getHost(),redisClusterNode.getPort(),keyPattern),
                            along.get());
                }
                return result;
            }
            long l = keyCountSingleton(keyPattern);
            Map<String,Long> result = new HashMap<>(1);
            result.put("localhost",l);
            return result;
    }

    public long keyCountSingleton(String keyPattern) {

        RedisConnection connection = getConnection();
        final Long  dbSize= connection.dbSize();
        if(!Objects.isNull(dbSize)){
            //根据当前数据库的大小，尽量在指定的循环次数中结束
            RedisScript<List<Object>> redisScript = new DefaultRedisScript(" local res = redis.call('SCAN',KEYS[1],'MATCH',KEYS[2],'COUNT',KEYS[3]); return {res[1],#res[2]} ", List.class);
            // 获得每次查询的COUNT数量，即redis查询的位数
            String scriptAsString = redisScript.getScriptAsString();
            final Integer[] args = new Integer[]{};
            final List<Boolean> booleans = getConnection().scriptExists(scriptAsString);
            if(!Objects.isNull(booleans)&&!booleans.isEmpty()&&!booleans.get(0)){
                connection.scriptLoad(scriptAsString.getBytes(StandardCharsets.UTF_8));
            }
            // 如果总数据量很小，就没必要每次都循环,每1w条，增加一次循环次数,阈值
            int loopCount = new Double((Math.floorDiv(dbSize, 10_000) * 10)*0.6).intValue()+1;
            final long count = Math.floorDiv(dbSize, loopCount);
            final String sha = redisScript.getSha1();
            Object[] keys = null; // help gc
            long keySize = 0;
            long cursor = 0;
            do{
                keys = new Object[]{Long.toString(cursor),keyPattern,Long.toString(count)};
                // 是否也要考虑分布式的问题

                List<Object> res = connection.evalSha(sha, ReturnType.fromJavaType(List.class),
                        keys.length, getExecutor().convertKeysAndArgs(keys, args));
                if(Objects.isNull(res)||res.isEmpty()) break;
                cursor =Long.parseLong(new String((byte[]) res.get(0)));
                keySize += (Long) res.get(1);
                // 当游标归0，标识遍历结束
            }while (cursor>0);
            return keySize;
        }
        return 0;
    }

    /**
     * 获得列表key的数量
     *
     */
    public long keyCount(String keyPattern){
        // 获取当前数据库的总大小
        RedisScript<Long> dbSizeScript = new DefaultRedisScript<>("return redis.call('dbsize')",Long.class);
        final Long  dbSize= getRedis().execute(dbSizeScript, Collections.emptyList(), "");
        if(!Objects.isNull(dbSize)){
            //根据当前数据库的大小，尽量在指定的循环次数中结束
            RedisScript<List<Object>> redisScript = new DefaultRedisScript(" local res = redis.call('SCAN',KEYS[1],'MATCH',KEYS[2],'COUNT',KEYS[3]); return {res[1],#res[2]} ", List.class);
            // 获得每次查询的COUNT数量，即redis查询的位数
            RedisConnection connection = getConnection();
                final List<Boolean> booleans = connection.scriptExists(redisScript.getScriptAsString());
                if(!Objects.isNull(booleans)&&!booleans.isEmpty()&&!booleans.get(0)){
                    connection.scriptLoad(redisScript.getScriptAsString().getBytes(StandardCharsets.UTF_8));
                }
                // 如果总数据量很小，就没必要每次都循环,每1w条，增加一次循环次数,阈值
                int loopCount = new Double((Math.floorDiv(dbSize, 10_000) * 10)*0.6).intValue()+1;
                final long count = Math.floorDiv(dbSize, loopCount);
                final String sha = redisScript.getSha1();
                final Integer[] args = new Integer[]{};
                Object[] keys = null; // help gc
                long keySize = 0;
                long cursor = 0;
                do{
                    keys = new Object[]{Long.toString(cursor),keyPattern,Long.toString(count)};
                    List<Object> res = connection.evalSha(sha, ReturnType.fromJavaType(List.class),
                            keys.length, getExecutor().convertKeysAndArgs(keys, args));
                    if(Objects.isNull(res)||res.isEmpty()) break;
                    cursor =Long.parseLong(new String((byte[]) res.get(0)));
                    keySize += (Long) res.get(1);
                    // 当游标归0，标识遍历结束
                }while (cursor>0);
                return keySize;
        }
        return 0;
    }

    @Override
    public void close() throws IOException {
        // 貌似当connection 为redisClusterConnection的时候，无法正常关闭
        RedisConnectionUtils.releaseConnection(getConnection(), getRedis().getRequiredConnectionFactory());
    }


    static class MyScriptExecutor extends DefaultScriptExecutor<Object> {
        private RedisTemplate redisTemplate;
        public MyScriptExecutor(RedisTemplate template) {
            super(template);
            this.redisTemplate = template;
        }

        public byte[][] convertKeysAndArgs(Object[] keys, Object[] args) {
            return super.keysAndArgs(this.redisTemplate.getValueSerializer(), Arrays.asList(keys), args);
        }
    }
}
